package com.zr.java8.thread;

import java.util.LinkedList;
import java.util.Queue;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

public class ProducerConsumerWithCondition {

    public static void main(String[] args) {
        // 创建任务缓冲区
        TaskBuffer buffer = new TaskBuffer(5);

        // 创建生产者线程
        Thread producer1 = new Thread(() -> {
            for (int i = 1; i <= 10; i++) {
                try {
                    buffer.produce("交易任务-" + i);
                    Thread.sleep(100); // 模拟任务生成时间
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }
        }, "生产者1");

        Thread producer2 = new Thread(() -> {
            for (int i = 11; i <= 20; i++) {
                try {
                    buffer.produce("订单任务-" + i);
                    Thread.sleep(150); // 模拟任务生成时间
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }
        }, "生产者2");

        // 创建消费者线程
        Thread consumer1 = new Thread(() -> {
            for (int i = 0; i < 15; i++) {
                try {
                    buffer.consume();
                    Thread.sleep(200); // 模拟任务处理时间
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }
        }, "消费者1");

        Thread consumer2 = new Thread(() -> {
            for (int i = 0; i < 15; i++) {
                try {
                    buffer.consume();
                    Thread.sleep(250); // 模拟任务处理时间
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }
        }, "消费者2");

        // 启动线程
        producer1.start();
        producer2.start();
        consumer1.start();
        consumer2.start();

        try {
            producer1.join();
            producer2.join();
            consumer1.join();
            consumer2.join();
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }

        System.out.println("所有任务处理完成!");
    }

    // 任务缓冲区类
    static class TaskBuffer {
        private final Queue<String> queue;  // 任务队列
        private final int capacity;        // 缓冲区容量
        private final ReentrantLock lock;  // 可重入锁
        private final Condition notFull;    // "缓冲区未满"条件
        private final Condition notEmpty;   // "缓冲区非空"条件

        public TaskBuffer(int capacity) {
            this.capacity = capacity;
            this.queue = new LinkedList<>();
            this.lock = new ReentrantLock(true); // 公平锁
            this.notFull = lock.newCondition();
            this.notEmpty = lock.newCondition();
        }

        // 生产者方法：添加任务
        public void produce(String task) throws InterruptedException {
            lock.lock(); // 获取锁
            try {
                // 当缓冲区满时，等待"notFull"条件
                while (queue.size() == capacity) {
                    System.out.println(Thread.currentThread().getName() + ": 缓冲区已满，等待中...");
                    notFull.await(); // 释放锁并等待
                }

                // 添加任务到缓冲区
                queue.offer(task);
                System.out.println(Thread.currentThread().getName() + " 生产: " + task +
                        " | 缓冲区大小: " + queue.size());

                // 通知消费者缓冲区非空
                notEmpty.signalAll();
            } finally {
                lock.unlock(); // 确保锁被释放
            }
        }

        // 消费者方法：处理任务
        public void consume() throws InterruptedException {
            lock.lock(); // 获取锁
            try {
                // 当缓冲区为空时，等待"notEmpty"条件
                while (queue.isEmpty()) {
                    System.out.println(Thread.currentThread().getName() + ": 缓冲区为空，等待中...");
                    notEmpty.await(); // 释放锁并等待
                }

                // 从缓冲区取出任务
                String task = queue.poll();
                System.out.println(Thread.currentThread().getName() + " 消费: " + task +
                        " | 缓冲区大小: " + queue.size());

                // 通知生产者缓冲区未满
                notFull.signalAll();
            } finally {
                lock.unlock(); // 确保锁被释放
            }
        }
    }
}